package com.cyf.mq.consumer;

import com.cyf.mq.constant.MqConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.time.LocalDateTime;
import java.util.List;

/**
 * @author chenyifan
 * @create 2025-02-20 11:00
 */
@Slf4j
public class RetryConsumer {
    /**
     * 重试时间间隔
     * 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * 默认重试16次
     * 1. 能否自定义重试次数
     * 2. 如果重试了16次（并发模式）顺序模式下（int最大值次）都是失败的？ 认为该消息是一个死信消息 则会放在一个死信主题中去。
     *    主题的名称：%DLQ%消费组的名字
     * 3. 当消息处理失败的时候 该如何正确的处理？
     * -----------
     * 重试的次数一般 5次
     * @throws Exception
     */
    @Test
    public void retryConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConstant.MESSAGE_CONSUMER_GROUP_A);
        consumer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
        consumer.subscribe(MqConstant.MESSAGE_TOPIC_King, "*");

        // 设置重试次数
        consumer.setMaxReconsumeTimes(1);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    log.info("consumer key:{}, msg:{}", msg.getKeys(), new String(msg.getBody()));
                }
                // 报错了 返回null 返回RECONSUME_LATER  都会重试
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        consumer.start();

        System.in.read();
    }


    @Test
    public void retryDeadConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConstant.MESSAGE_DEAD_CONSUMER_GROUP);
        consumer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);

        consumer.subscribe("%DLQ%" + MqConstant.MESSAGE_CONSUMER_GROUP_A, "*");

        // 设定重试次数
        consumer.setMaxReconsumeTimes(2);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    log.info("msg:{} 记录到特别的位置 文件 mysql 通知人工处理", new String(msg.getBody()));
                }
                // 业务报错了 返回null 返回RECONSUME_LATER  都会重试
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.in.read();
    }
}
